package com.dmp.tags

import com.dmp.beans.Logs
import com.dmp.utils.Utils
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 7）上下文标签：将数据打上上述6类标签，并根据【用户ID】进行当前文件的合并，数据保存格式为
  *
  * 运行参数为：
  * G:\光环国际大数据开发班\大数据最后阶段-项目\22-dmp项目\资料\data.txt G:\光环国际大数据开发班\大数据最后阶段-项目\22-dmp项目\资料\appmapping.txt G:\光环国际大数据开发班\大数据最后阶段-项目\22-dmp项目\资料\device_mapping.txt xx
  */
object TagsContext {
  def main(args: Array[String]): Unit = {
    //判断参数
    if (args.length < 4) {
      println(
        """
          |com.dmp.tags.TagsContext
          |<inputLogPath> 输入的日志文件路径
          |<appMappingPath> app映射文件路径 appmapping.txt
          |<deviceMappingPath>设备的映射文件路径 device_mapping.txt
          |<outputPath> 输出的结果文件存储
        """.stripMargin)
      System.exit(0)
    }
    //接收参数
    val Array(inputPath, appPath, devicePath, outputPath) = args
    //初始化对象
    val conf = new SparkConf()
    conf.setAppName(s"${this.getClass.getSimpleName}")
    conf.setMaster("local")
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    conf.registerKryoClasses(Array(classOf[Logs]))
    //初始化程序入口
    val sc = new SparkContext(conf)
    //生成 APP映射（AppId->AppName） 广播 变量
    val appMap: Map[String, String] = sc.textFile(appPath).flatMap(line => {
      var map = Map[String, String]()
      val fields = line.split("\t")
      // 4 是APP唯一表示号
      if (fields.length > 4) {
        map += (fields(4) -> fields(1))
      }
      map
    }).collect().toMap

    val appMapBroadcast = sc.broadcast(appMap)

    //生成设备映射广播变量
    val deviceMap: Map[String, String] = sc.textFile(devicePath).map(line => {
      var map = Map[String, String]()
      val fields = line.split("\t")
      if (fields.length > 1) {
        map += (fields(0) -> fields(1))
      }
      map
    }).collect.flatten.toMap

    val deviceMapBroadcast = sc.broadcast(deviceMap)

    //进行打标签
    sc.textFile(inputPath).map(line => {
      val log = Logs.line2Log(line)
      //1）广告位类型 打标签返回Map类型
      val localTag: Map[String, Int] = Tags4Local.makeTags(log)
      //2）APP名称 打标签返回Map类型
      val appTag = Tags4App.makeTags(log, appMapBroadcast.value)
      //3）渠道 打标签返回Map类型
      val channelTag = Tags4Channel.makeTags(log)
      //4）设备：操作系统|联网方式|运营商 打标签返回Map类型
      val deviceTag = Tags4Device.makeTags(log, deviceMapBroadcast.value)
      //5）关键词 打标签返回Map类型
      val keyWordsTag = Tags4KeyWords.makeTags(log)
      //6）地域标签 打标签返回Map类型
      val areaTag = Tags4Area.makeTags(log)
      //getNotEmptyID(log)获取到一个ID，返回值是个Option，有可能有，有可能没有，如果没有打出来的标签是没有意义的，给个默认值""
      val userid: String = getNotEmptyID(log).getOrElse("")

      /**
        * (localTag ++  appTag ++ channelTag  ++ deviceTag ++ keyWordsTag ++ areaTag) 表示所有的Map都合并到同一个Map中了
        * val map3 = Map("Kmala"->1)
        * val map4 = Map("APPaiyiqi"->1)
        * map3 ++ map4 就变成下面的样子了
        * Map("Kmala"->1,"APPaiyiqi"->1)
        */
      (userid, (localTag ++ appTag ++ channelTag ++ deviceTag ++ keyWordsTag ++ areaTag).toList)
    })
      //过滤，如果userid不等于空，我们就要这个数据
      .filter(!_._1.toString.equals(""))
      .reduceByKey {
        case (list1, list2) => {
          /**
            * var b=Map("asd"->1)
            * var a=List("qwe",1)
            * b++a 后就是下面的形式
            * List((asd,1),qwe,1)
            */
          (list1 ++ list2).groupBy(_._1)
            .map {
              case (k, list) => {
                (k, list.map(t => t._2).sum)
              }
            }.toList
        }
      }.foreach(tuple => {
      println(tuple._1 + "->" + tuple._2.mkString("\t"))

    })
    sc.stop()
  }

  // 获取用户唯一不为空的ID（不需要我们管）
  def getNotEmptyID(log: Logs): Option[String] = {
    log match {
      case v if v.imei.nonEmpty => Some("IMEI:" + Utils.formatIMEID(v.imei))
      case v if v.imeimd5.nonEmpty => Some("IMEIMD5:" + v.imeimd5.toUpperCase)
      case v if v.imeisha1.nonEmpty => Some("IMEISHA1:" + v.imeisha1.toUpperCase)

      case v if v.androidid.nonEmpty => Some("ANDROIDID:" + v.androidid.toUpperCase)
      case v if v.androididmd5.nonEmpty => Some("ANDROIDIDMD5:" + v.androididmd5.toUpperCase)
      case v if v.androididsha1.nonEmpty => Some("ANDROIDIDSHA1:" + v.androididsha1.toUpperCase)

      case v if v.mac.nonEmpty => Some("MAC:" + v.mac.replaceAll(":|-", "").toUpperCase)
      case v if v.macmd5.nonEmpty => Some("MACMD5:" + v.macmd5.toUpperCase)
      case v if v.macsha1.nonEmpty => Some("MACSHA1:" + v.macsha1.toUpperCase)

      case v if v.idfa.nonEmpty => Some("IDFA:" + v.idfa.replaceAll(":|-", "").toUpperCase)
      case v if v.idfamd5.nonEmpty => Some("IDFAMD5:" + v.idfamd5.toUpperCase)
      case v if v.idfasha1.nonEmpty => Some("IDFASHA1:" + v.idfasha1.toUpperCase)

      case v if v.openudid.nonEmpty => Some("OPENUDID:" + v.openudid.toUpperCase)
      case v if v.openudidmd5.nonEmpty => Some("OPENDUIDMD5:" + v.openudidmd5.toUpperCase)
      case v if v.openudidsha1.nonEmpty => Some("OPENUDIDSHA1:" + v.openudidsha1.toUpperCase)

      case _ => None
    }

  }

}
